package cn.itcast.batch.task;

import cn.itcast.batch.TotalDataRate;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.io.jdbc.JDBCOutputFormat;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.types.Row;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

//定义按照汇总统计数据准确率计算任务
public class TotalDataRateTask {
    private static final Logger logger = LoggerFactory.getLogger(TotalDataRateTask.class);
    public static void main(String[] args) {
        /**
         * 1.初始化flink批处理的运行环境
         * 2.实例化计算汇总统计数据准确率对象
         * 3.读取hive的数据
         * 4.将查询的结果转换为DataSource对象（计算准确率）
         * 5.得到jdbcOutFormat对象
         * 6.将计算好的数据指标写入到mysql中
         * 7.执行任务
         */

        //todo 1.初始化flink批处理的运行环境
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        //todo 2.实例化计算汇总统计数据准确率对象
        TotalDataRate totalDataRate = new TotalDataRate();

        //todo 3.读取hive的数据
        DataSource<Row> hiveDataSource = env.createInput(totalDataRate.getHiveJDBCInputFormat());

        //todo 4.将查询的结果转换为DataSource对象（计算准确率）
        DataSet<Row> hiveDataSet = totalDataRate.convertHiveDataSource(hiveDataSource);

        //todo 5.得到jdbcOutFormat对象
        JDBCOutputFormat mysqlJDBCOutputFormat = totalDataRate.getMysqlJDBCOutputFormat();

        //todo 6.将计算好的数据指标写入到mysql中
        hiveDataSet.output(mysqlJDBCOutputFormat);

        //todo 7.执行任务
        try {
            env.execute();
        } catch (Exception e) {
            logger.info(e.getMessage());
            e.printStackTrace();
        }
    }
}
